package com.bang.springcloud.report.util;

import com.alibaba.fastjson.JSON;
import com.bang.framework.mq.service.ReliableDeliveryService;
import com.bang.springcloud.report.entity.constants.Constants;
import com.bang.springcloud.report.mapper.BrokerMessageLogMapper;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.AmqpException;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessagePostProcessor;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import org.springframework.util.StringUtils;

import java.time.LocalDateTime;

/**
 * @ClassName: OrderSendUtils
 * @author: BANG
 * @Date: 2019/12/10 11:02
 * @Version: 1.0
 * @Modified By:
 * @Desc: TODO
 */
@Slf4j
@Component
public class OrderSendUtils implements ReliableDeliveryService {


    @Autowired
    private RabbitTemplate rabbitTemplate;

    @Autowired
    private BrokerMessageLogMapper brokerMessageLogMapper;

    @Override
    public <T> void send(T message, CorrelationData correlationData, String routerKey) throws Exception {
        log.info("写入消息队列：{}，消息内容：{}", routerKey, JSON.toJSONString(message));
        //设置回调为当前类对象
        rabbitTemplate.setConfirmCallback(this);
        rabbitTemplate.setReturnCallback(this);
        //发送消息到消息队列
        rabbitTemplate.convertAndSend(routerKey, message, correlationData);
    }

    @Override
    public <T> void send(T message, CorrelationData correlationData, String exchange, String routerKey) throws Exception {
        log.info("写入消息队列：{}，消息内容：{}", routerKey, JSON.toJSONString(message));
        //设置回调为当前类对象
        rabbitTemplate.setConfirmCallback(this);
        rabbitTemplate.setReturnCallback(this);
        //构建回调id为uuid
        //发送消息到消息队列
        rabbitTemplate.convertAndSend(exchange, routerKey, message, correlationData);
    }

    @Override
    public <T> void sendTtlMessage(T messageContent, CorrelationData correlationData, String exchange, String routerKey, long delayTimes) throws Exception {
        //设置回调为当前类对象
        rabbitTemplate.setConfirmCallback(this);
        rabbitTemplate.setReturnCallback(this);

        //构建回调id为uuid
        //构建回调id为uuid
        if (!StringUtils.isEmpty(exchange)) {
            log.info("延迟：{}毫秒写入消息队列：{}，消息内容：{}", delayTimes, routerKey, JSON.toJSONString(messageContent));
            // 执行发送消息到指定队列
            rabbitTemplate.convertAndSend(exchange, routerKey, messageContent, message -> {
                message.getMessageProperties().setExpiration(String.valueOf(delayTimes));
                return message;
            }, correlationData);
        } else {
            log.error("未找到队列消息：{}，所属的交换机", exchange);
        }
    }

    @Override
    public void confirm(CorrelationData correlationData, boolean ack, String cause) {
        log.error("correlationData: {}", correlationData);
        String messageId = correlationData.getId();
        if (ack) {
            //如果confirm返回成功 则进行更新
            brokerMessageLogMapper.changeBrokerMessageLogStatus(messageId, Constants.ORDER_SEND_SUCCESS, LocalDateTime.now());
        } else {
            //失败则进行具体的后续操作:重试 或者补偿等手段
            log.error("异常处理....");
        }
    }

    @Override
    public void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey) {
        log.error("return exchange: {}, routingKey: {}, replyCode: {}, replyText: {}", message, routingKey, replyCode, replyText);
    }
}
